package com.tour.dwd

import java.awt.geom.Point2D
import java.text.SimpleDateFormat
import java.util.Date

import com.shujia.common.{Constants, SparkTool}
import com.shujia.common.grid.Grid
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import scala.collection.mutable.ListBuffer

object StayPointApp extends SparkTool {

  override def run(spark: SparkSession): Unit = {

    import spark.implicits._

    //1、读取融合表
    /**
      *
      * mdn string comment '手机号码'
      * ,start_time string comment '业务时间'
      * ,county_id string comment '区县编码'
      * ,longi string comment '经度'
      * ,lati string comment '纬度'
      * ,bsid string comment '基站标识'
      * ,grid_id string comment '网格号'
      * ,biz_type string comment '业务类型'
      * ,event_type string comment '事件类型'
      * ,data_source string comment '数据源'
      */

    val mergeLocation: DataFrame = spark.sql(s"select mdn,start_time,county_id,grid_id from ${Constants.DWD_DATABASE_NAME}.${Constants.MERGELOCATION_TABLE_NAME}  where day_id=$day_id")

    /**
      * 按照时间顺序将同一个人在同一个网格中的多条数据合并成一条
      *
      */

    //将DF转换成DS

    val kvRDD: RDD[(String, (String, String, String, String))] = mergeLocation.rdd.map(row => {
      val mdn: String = row.getAs[String]("mdn")
      val start_time: String = row.getAs[String]("start_time")
      val county_id: String = row.getAs[String]("county_id")
      val grid_id: String = row.getAs[String]("grid_id")

      //结束时间
      val endTime: String = start_time.split(",")(0)
      //开始时间
      val startTime: String = start_time.split(",")(1)

      //以手机号作为key
      (mdn, (county_id, grid_id, startTime, endTime))
    })

    //按手机号分组,将同一个人一天的数据分到同一个组内
    val groupRDD: RDD[(String, Iterable[(String, String, String, String)])] = kvRDD.groupByKey()
    //使用flatMap 返回值是一个ListBuffer,会自动展开
    val reslutPointRDD: RDD[(String, String, String, String, String)] = groupRDD.flatMap {
      case (mdn: String, points: Iterable[(String, String, String, String)]) =>

        //使用集合保存最终输出结果
        val resultPoints = new ListBuffer[(String, String, String, String, String)]

        val pointList: List[(String, String, String, String)] = points.toList

        //1、按照时间排序
        val sortPointsList: List[(String, String, String, String)] = pointList.sortBy(_._4)

        //2、先获取第一个点
        val headPoint: (String, String, String, String) = sortPointsList.head

        var headCountyId: String = headPoint._1
        var headGridId: String = headPoint._2
        var headStartTime: String = headPoint._3
        var headEndTime: String = headPoint._4

        //3、循环后面的点和前一个点进行对比
        val tailPoints: List[(String, String, String, String)] = sortPointsList.tail

        tailPoints.foreach {
          case (county_id: String, grid_id: String, startTime: String, endTime: String) =>
            //用当前点的网格编号和上一个点的网格编号进行对比
            if (headGridId.equals(grid_id)) {
              headEndTime = endTime
            } else {

              //将前面几个相同的点保存起来
              resultPoints.+=((mdn, headCountyId, headGridId, headStartTime, headEndTime))

              //切换下一个组
              headGridId = grid_id
              headCountyId = county_id
              headStartTime = startTime
              headEndTime = endTime
            }
        }

        //处理最后一组
        resultPoints.+=((mdn, headCountyId, headGridId, headStartTime, headEndTime))


        //返回数据
        resultPoints

    }

    /**
      * mdn string comment '用户手机号码'
      * ,longi string comment '网格中心点经度'
      * ,lati string comment '网格中心点纬度'
      * ,grid_id string comment '停留点所在电信内部网格号'
      * ,county_id string comment '停留点区县'
      * ,duration string comment '机主在停留点停留的时间长度（分钟）,lTime-eTime'
      * ,grid_first_time string comment '网格第一个记录位置点时间（秒级）'
      * ,grid_last_time string comment '网格最后一个记录位置点时间（秒级）'
      */


    //计算网格中心的的经纬度和用户在网格中的停留时间
    val stayPointRDD: RDD[(String, String, String, String, String, String, String, String)] = reslutPointRDD.map {
      case (mdn: String, county_id: String, grid_id: String, startTime: String, endTime: String) =>
        //1、获取网格中心点经纬度
        val point: Point2D.Double = Grid.getCenter(grid_id.toLong)
        val longi: Double = point.getX
        val lati: Double = point.getY

        //2、计算停留时间
        val format = new SimpleDateFormat("yyyyMMddHHmmss")

        val startDate: Date = format.parse(startTime)
        val endDate: Date = format.parse(endTime)

        //停留时间
        val duration: Double = (endDate.getTime - startDate.getTime) / 60000.0

        (mdn, longi.formatted("%.4f"), lati.formatted("%.4f"), grid_id, county_id, duration.formatted("%.4f"), startTime, endTime)
    }


    //保存数据
    stayPointRDD
      .toDF()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"${Constants.STAYPOINT_PATH_NAME}day_id=$day_id")

    //增加分区
    //需要先创建表
    spark.sql(s"alter table ${Constants.DWD_DATABASE_NAME}.${Constants.STAYPOINT_TABLE_NAME} add if not exists partition(day_id='$day_id') ")


    /**
      *
      * spark-submit --master yarn-client --class com.tour.dwd.StayPointApp --executor-memory 4G --executor-core
      * s 2  --num-executors 2 --jars common-1.0-SNAPSHOT.jar  dwd-1.0-SNAPSHOT.jar 20180503
      *
      */

  }
}
